package com.liuwenlin.zookeeper.zkclient.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 * 实现 Zookeeper 分布式锁
 * @author 276834
 */
public class ZookeeperLock implements Lock, Watcher, AsyncCallback.StringCallback, AsyncCallback.Children2Callback, AsyncCallback.StatCallback {

    /**
     * zk客户端
     */
    private ZooKeeper zkClient;

    private CountDownLatch cdl = new CountDownLatch(1);

    /**
     * 线程名称
     */
    private String threadName;

    private String pathName;

    public String getThreadName() {
        return threadName;
    }

    public void setThreadName(String threadName) {
        this.threadName = threadName;
    }

    public String getPathName() {
        return pathName;
    }

    public void setPathName(String pathName) {
        this.pathName = pathName;
    }

    public ZooKeeper getZkClient() {
        return zkClient;
    }

    public void setZkClient(ZooKeeper zkClient) {
        this.zkClient = zkClient;
    }

    @Override
    public void lock() {

    }

    @Override
    public void lockInterruptibly() throws InterruptedException {

    }

    @Override
    public boolean tryLock() {
        System.out.println(threadName + " is creating lock...");
        try {
            //锁获取起始步骤，创建锁节点，加入异步回调获取节点名称的接口实现AsyncCallback.StringCallback
            //注意：create()方法的第二个参数byte[]是设置到zk连接的根节点上的
            zkClient.create("/lock", threadName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL, this, "createLock");
            cdl.await();
            return true;
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return false;
    }

    @Override
    public void unlock() {
        try {
            //这里的删除节点操作，已经在监控前一个子节点的异步处理方法中注册了watcher
            zkClient.delete(pathName, -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }

    @Override
    public Condition newCondition() {
        return null;
    }

    @Override
    public void process(WatchedEvent event) {
        switch(event.getType()){
            case None:
                break;
            case NodeCreated:
                break;
            case NodeDeleted:
                //不管哪一个节点正常或者异常退出（即request session结束时），当前线程都再一次获取根节点的子节点情况，继续后续操作（可以响应中断）
                zkClient.getChildren("/",false,this ,"getChildren");
                break;
            case NodeDataChanged:
                break;
            case NodeChildrenChanged:
                break;
            default:
                System.out.println("Not matched type.");
        }
    }

    @Override
    public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {
        //一定能看到自己前边的。。

//        System.out.println(threadName+"look locks.....");
//        for (String child : children) {
//            System.out.println(child);
//        }

        Collections.sort(children);
        int i = children.indexOf(pathName.substring(1));
        //是不是第一个
        if(i == 0){
            //yes
            System.out.println(threadName +" i am first....");
            try {
                //这里在根目录设置值，为了保证并发线程都能及时响应，获取根目录下的临时顺序节点
                zkClient.setData("/", threadName.getBytes(),-1);

            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                cdl.countDown();
            }
        } else {
            //no 不是第一个子节点， 这里就监控前一个子节点，需要传入对应节点watcher，以监听前一个节点stat的变化，然后获取后续的操作
            zkClient.exists("/"+children.get(i-1),this,this,"watchPreNode");
        }

    }

    @Override
    public void processResult(int rc, String path, Object ctx, Stat stat) {
        //todo 当前锁队列节点的状态信息暂时可以不用处理
        //如果自己没有watch到，说明上一个节点突然挂掉了。那就需要重新判断自己的位置
        if(KeeperException.Code.NONODE.intValue() == rc){
            zkClient.getChildren("/", false, this, "getChildren");
        }
    }

    @Override
    public void processResult(int rc, String path, Object ctx, String name) {
        if(name != null){
            //创建node时返回的节点名称记录下来
            System.out.println(threadName  +"  create node : " +  name);
            pathName = name ;
            //这里就是发起获取根目录“所有子节点”的异步回调的方法getChildren(),用于回调AsyncCallback.Children2Callback接口的实现
            zkClient.getChildren("/",false,this ,"getChildren");
        }

    }
}
